package hust.rpc.handler;

import hust.rpc.pojo.Node;
import hust.rpc.utils.CommonUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * 注册中心处理器
 */
@Slf4j
@ChannelHandler.Sharable
public class RegistryHandler extends RpcInvokeHandler {

    // 定时线程池,用于定时运行任务(将提供者上下线信息发送给消费者)
    private static final ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);

    // 消费者节点, <通道Channel, 上次更新提供者信息的时间戳>
    private static final Map<Channel, Long> consumerNodes = new ConcurrentHashMap<>();

    // 更新的提供者节点信息(上/下线、更新时间戳、绑定二元组、提供的服务集合(上线才有))
    private static final ConcurrentSkipListSet<Node> changedProviderNodes = new ConcurrentSkipListSet<>((o1, o2) -> {
        if (o1.getTimestamp() == o2.getTimestamp()) return 0;
        return o1.getTimestamp() < o2.getTimestamp() ? 1 : -1;
    });

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 单线程线程池处理心跳连接
        super.channelActive(ctx);
    }

    /**
     * 更新服务提供者上/下线信息
     *
     * @param node 提供者节点信息
     */
    public void updateProviders(Node node) {
        log.info("服务提供者{}在{}时{}线", node.getHost(), node.getTimestamp(), node.isAdd() ? '上' : '下');
        changedProviderNodes.add(node);
    }

    /**
     * 更新消费者上/下线信息
     *
     * @param channel 通道信息
     * @param online  是否上线
     */
    public void updateConsumers(Channel channel, boolean online) {
        log.info("服务消费者{}在{}时{}线", CommonUtils.channelToHost(channel), System.currentTimeMillis(), online ? '上' : '下');
        if (online) {
            consumerNodes.put(channel, System.currentTimeMillis());
        } else {
            consumerNodes.remove(channel);
        }
    }
}
